package com.lyloou.boot;


import com.lyloou.client.RPCConsumer;
import com.lyloou.server.IUserService;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ConsumerBoot {
    private Map<String, IUserService> map = new ConcurrentHashMap<>();

    public static void main(String[] args) throws Exception {
        ConsumerBoot consumerBoot = new ConsumerBoot();
        consumerBoot.start();

        consumerBoot.sendData();
    }

    private void start() throws Exception {
        RetryPolicy exponentialBackoffRetry = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("192.168.1.142:2281")
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .retryPolicy(exponentialBackoffRetry)
                .namespace("base")  // 独立的命名空间 /base
                .build();
        client.start();

        String path = "/rpc";
        PathChildrenCacheListener pcl = (curatorFramework, pathChildrenCacheEvent) -> {
            System.out.println("服务节点发生了变化..." + pathChildrenCacheEvent.getType());
            System.out.println(pathChildrenCacheEvent.getData().getPath());
            try {
                List<String> newChildren = client.getChildren().forPath(path);
                Set<String> oldChildren = map.keySet();

                // 3. 某个服务端下线后，Zookeeper注册列表会自动剔除下线的服务端节点，客户端与下线的服务端断开连接
                List<String> deletedChildren = new ArrayList<>(oldChildren);
                deletedChildren.removeAll(newChildren);
                for (String deletedChild : deletedChildren) {
                    System.out.println("移除了：" + deletedChild);
                    map.remove(deletedChild);
                }

                // 4. 服务端重新上线，客户端能感知到，并且与重新上线的服务端重新建立连接
                List<String> createChildren = new ArrayList<>(newChildren);
                createChildren.removeAll(oldChildren);
                refreshMap(client, path, createChildren);
            } catch (Exception e) {
                e.printStackTrace();
            }
        };

        PathChildrenCache cache = new PathChildrenCache(client, path, false);
        cache.start();
        cache.getListenable().addListener(pcl);


        List<String> children = client.getChildren().forPath(path);
        refreshMap(client, path, children);
    }

    private void refreshMap(CuratorFramework client, String path, List<String> createChildren) throws Exception {
        for (String createChild : createChildren) {
            String childPath = path + "/" + createChild;
            String[] data = new String(client.getData().forPath(childPath)).split(":");
            String host = data[0];
            String port = data[1];
            IUserService service = (IUserService) new RPCConsumer().createProxy(IUserService.class, host, Integer.parseInt(port));
            map.put(createChild, service);
            System.out.println("新增节点：" + createChild);
        }
    }

    //1.创建一个线程池对象  -- 它要处理我们自定义事件
    private static ExecutorService executorService =
            Executors.newCachedThreadPool();

    private void sendData() {
        //2.循环给服务器写数据
        while (true) {
            System.out.println("\nrunning..." + map.keySet());
            for (String child : map.keySet()) {
                IUserService service = map.get(child);
                executorService.submit(() -> {
                    String result = service.sayHello("client say hello to " + child);
                    System.out.println("result from server " + child + ": " + result);
                });
            }

            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }
}
